import time
import yaml

from canal.client import Client
from canal.protocol import EntryProtocol_pb2


file_path = "../etc/config.yaml"
with open(file_path, 'r', encoding='utf-8') as file:
    config = yaml.safe_load(file)
    canal_config = config.get('canal')
    if not canal_config:
        raise ValueError("Canal configuration not found in the YAML file.")


host = canal_config.get('address')
port = canal_config.get('port')
username = canal_config.get('username').encode()
password = canal_config.get('password').encode()
destination = canal_config.get('destination').encode()
filters = canal_config.get('filter').encode()
client_id = canal_config.get('clientId').encode()

# 配置 Canal 连接信息
client = Client()
client.connect(host=host, port=port)
client.check_valid(username=username, password=password)
client.subscribe(client_id=client_id, destination=destination, filter=filters)


try:
    while True:
        message = client.get(100)
        entries = message['entries']
        for entry in entries:
            if entry.entryType == EntryProtocol_pb2.EntryType.ROWDATA:
                row_change = EntryProtocol_pb2.RowChange()
                row_change.MergeFromString(entry.storeValue)
                event_type = row_change.eventType
                header = entry.header
                database = header.schemaName
                table = header.tableName

                print(f"Database: {database}, Table: {table}, Event Type: {event_type}")

                for row in row_change.rowDatas:
                    if event_type == EntryProtocol_pb2.EventType.INSERT:
                        columns = [column.name + ": " + column.value for column in row.afterColumns]
                        print("Insert event:", columns)
                    elif event_type == EntryProtocol_pb2.EventType.UPDATE:
                        columns = [column.name + ": " + column.value for column in row.afterColumns]
                        print("Update event:", columns)
                    elif event_type == EntryProtocol_pb2.EventType.DELETE:
                        columns = [column.name + ": " + column.value for column in row.beforeColumns]
                        print("Delete event:", columns)

        time.sleep(1)

except KeyboardInterrupt:
    print("Interrupted by user. Closing connection...")
finally:
    client.disconnect()
